package com.lg.flink.db;

import com.lg.flink.bean.WC;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;

public class JdbcCount {
    public static void main(String[] args) throws Exception {
        //scala代码
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);
        env.enableCheckpointing(5000);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        DataStreamSource dataStream =  env.addSource(new JdbcReader());//，读取mysql数据，获取dataStream后可以做逻辑处理，这里没有做
//        dataStream.keyBy(0)
//                .sum(1).print();
        Table table = tEnv.fromDataStream(dataStream);

        Table filtered = table
                .groupBy("word")
                .select("word, frequency.sum as frequency")
                ;

        tEnv.toRetractStream(filtered,WC.class).writeAsCsv("E:\\a.txt");
        env.execute("flink mysql demo");//运行程序

    }
}
